package com.zbj.storm.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * KafkaConsumer
 *
 * @author weigang
 * @create 2019-09-16
 **/
public class KafkaConsumer {

    private final org.apache.kafka.clients.consumer.KafkaConsumer consumer;

    private KafkaConsumer() {
        //Properties properties = new Properties();
        //// zookeeper 配置
        //properties.put("zookeeper.connect", "172.31.15.175:218");
        //
        //// 代表一个消费组
        //properties.put("group.id", "zbj-group");
        //
        //// zk连接超时
        //properties.put("zookeeper.session.timeout.ms", "4000");
        //properties.put("zookeeper.sync.time.ms", "200");
        //properties.put("auto.commit.interval.ms", "1000");
        //properties.put("auto.offset.reset", "smallest");
        //
        //// 序列化类
        //properties.put("", "");
        //
        //consumer = new org.apache.kafka.clients.consumer.KafkaConsumer(properties);

        Properties props = new Properties();
        //props.put("zookeeper.connect", "172.31.15.175:2181");
        props.put("bootstrap.servers", "172.31.15.175:9092");
        props.put("group.id", "zbj-group");
        props.put("max.poll.records", 100);
        props.put("enable.auto.commit", "false");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        //props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new org.apache.kafka.clients.consumer.KafkaConsumer(props);
    }

    void consume() {

        consumer.subscribe(Arrays.asList(KafkaProducer.TOPIC));
        final int minBatchSize = 200;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                buffer.add(record);
            }
            if (buffer.size() >= minBatchSize) {
                // 入库操作
                //insertIntoDb(buffer);
                for (ConsumerRecord<String, String> record : buffer) {
                    System.out.println(record.key() + "-> " + record.value());
                }

                consumer.commitSync();
                buffer.clear();
            }
        }
    }

    public static void main(String[] args) {
        new KafkaConsumer().consume();
    }
}